package com.atguigu.flink.streamapi.exec;

import com.atguigu.flink.pojo.OrderEvent;
import com.atguigu.flink.pojo.TxEvent;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;

/**
 * Created by Smexy on 2022/11/22
 *
 需求：
    在OrderLog.csv，有数据 OrderEvent
         34729,pay,sd76f87d6,1558430844
         OrderLog.csv:订单id,行为,txID,ts

    在ReceiptLog.csv，有数据TxEvent
         sd76f87d6,wechat,1558430845
         ReceiptLog.csv:txID,支付渠道, ts

    将OrderEvent的txId 和  TxEvent的txId进行比对，如果比对成功，就输出成功的订单Id
 */
public class Demo5_Order
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //消费两种不同的数据，获取两个流
        SingleOutputStreamOperator<OrderEvent> orderDS = env.readTextFile("data/OrderLog.csv")
                                                        .map(new MapFunction<String, OrderEvent>()
                                                        {
                                                            @Override
                                                            public OrderEvent map(String value) throws Exception {
                                                                String[] words = value.split(",");
                                                                return new OrderEvent(
                                                                    Long.valueOf(words[0]),
                                                                    words[1],
                                                                    words[2],
                                                                    Long.valueOf(words[3])
                                                                );
                                                            }
                                                        });


        SingleOutputStreamOperator<TxEvent> txDS = env.readTextFile("data/ReceiptLog.csv")
                                                            .map(new MapFunction<String, TxEvent>()
                                                            {
                                                                @Override
                                                                public TxEvent map(String value) throws Exception {
                                                                    String[] words = value.split(",");
                                                                    return new TxEvent(
                                                                        words[0],
                                                                        words[1],
                                                                        Long.valueOf(words[2])
                                                                    );
                                                                }
                                                            });


        orderDS.connect(txDS)
               .keyBy(OrderEvent::getTxId,TxEvent::getTxId)   //按照key进行分发，保证 txId 相同的OrderEvent和TxEvent 可以发送到下游的同一个 CoProcessFunction
               .process(new MyCoProcessFunction())
               .print().setParallelism(1);

        //比对两个流中

        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }


    }

    private static class MyCoProcessFunction extends KeyedCoProcessFunction<String, OrderEvent, TxEvent, String>
    {

        //成员变量声明两个集合，存储OrderEvent和TxEvent
        // key: txId ,value: orderId
        Map<String,Long> orderMap = new HashMap<>();

        //只存txEvent的txId
        HashSet<String> txEvent = new HashSet<>();

            @Override
            public void processElement1(OrderEvent value, Context ctx, Collector<String> out) throws Exception {
                    //先去对方的集合中找，看是否能找到，能找到，对账成功，输出。
                    if (txEvent.contains(value.getTxId())){
                        out.collect(value.getOrderId().toString());
                    }else{
                        //找不到，可能没有。可能 对方数据还没来，把自己写入集合，等待后续对方数据到达后，再处理
                        orderMap.put(value.getTxId(),value.getOrderId());
                    }


            }

            @Override
            public void processElement2(TxEvent value, Context ctx, Collector<String> out) throws Exception {

                Long orderId = orderMap.get(value.getTxId());

                if (orderId != null){
                    out.collect(orderId.toString());
                }else{
                    txEvent.add(value.getTxId());
                }

            }


    }
}
